<?php
namespace app\rabbitmq\consumer;

use Workerman\Worker;
use Workerman\Timer;
use PhpAmqpLib\Connection\AMQPStreamConnection;

abstract class BaseConsumer
{
    protected $connection;
    protected $channel;

    const DIRS = ['high', 'normal', 'low'];

    /**
     * 子类可以覆盖此方法，手动指定要消费的队列
     * 默认自动扫描 app/queue/mq 下的 high/normal/low 目录
     */
    protected function getQueues(): array
    {
        $baseDir = app_path() . '/rabbitmq/job';
        $queues = [];

        foreach (self::DIRS as $dir) {
            $path = $baseDir . '/' . $dir;
            if (!is_dir($path)) continue;
            foreach (glob($path . '/*.php') as $file) {
                $queues[] = basename($file, '.php'); // 文件名就是类名，也就是队列名
            }
        }
        return $queues;
    }

    protected function handleMessage($queue, $msg): void
    {
        $found = false;

        foreach (self::DIRS as $dir) {
            $class = "app\\rabbitmq\\job\\{$dir}\\" . $queue; // 直接用扫描文件名
            if (class_exists($class)) {
                (new $class())->consumeMq($msg);
                $found = true;
                break;
            }
        }

        if (!$found) {
            echo "[!] No handler found for queue={$queue}\n";
            $msg->ack();
        }
    }

    public function onWorkerStart(Worker $worker)
    {
        $queues = $this->getQueues();

        $config = config('plugin.rabbitmq.config');

        $this->connection = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['user'],
            $config['password'],
            $config['vhost']
        );

        $this->channel = $this->connection->channel();

        // 动态订阅队列
        foreach ($queues as $queue) {
            $this->channel->queue_declare($queue, false, true, false, false);

            $this->channel->basic_consume(
                $queue,
                '',
                false,
                false,
                false,
                false,
                function ($msg) use ($queue) {
                    $this->handleMessage($queue, $msg);
                }
            );
        }

        // 定时器驱动
        Timer::add(0.1, function () {
            if ($this->channel->is_consuming()) {
                $this->channel->wait(null, true);
            }
        });
    }

    public function onWorkerStop(Worker $worker)
    {
        if ($this->channel) {
            $this->channel->close();
        }
        if ($this->connection) {
            $this->connection->close();
        }
    }
}